package com.bw.gmall.realtime.app.dwd;

import com.bw.gmall.realtime.utils.MyKafkaUtil;
import com.bw.gmall.realtime.utils.MysqlUtil;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

public class DwdTradeRefundPaySuc {
    public static void main(String[] args) throws Exception {

        //todo 1.环境准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);
        //获取配置对象
        Configuration configuration = tenv.getConfig().getConfiguration();
//        //过期时间
//        tenv.getConfig().setIdleStateRetention(Duration.ofSeconds(5));
//        configuration.setString("table.exec.state.ttl","5 s");


        //todo 从kafka读取topic_db数据  封装为flink_sql类
        tenv.executeSql("create table `topic_db` ( " +
                "`database` string, " +
                "`table` string, " +
                "`type` string, " +
                "`data` map<string,string>, " +
                "`old` map<string,string>, " +
                "`proc_time`  as proctime(), " +
                "`ts` string " +
                ")"+ MyKafkaUtil.getKafkaDDL("topic_db","refund_pay_suc"));

        //todo 建立字典表
        tenv.executeSql(MysqlUtil.getBaseDicLookUpDDL());

        //TODO 5. 读取退款表数据，并筛选退款成功数据

        Table refundPayment = tenv.sqlQuery("select " +
                        "data['id'] id, " +
                        "data['order_id'] order_id, " +
                        "data['sku_id'] sku_id, " +
                        "data['payment_type'] payment_type, " +
                        "data['callback_time'] callback_time, " +
                        "data['total_amount'] total_amount, " +
                        "proc_time, " +
                        "ts " +
                        "from topic_db " +
                        "where `table` =  'refund_payment'" +
                        "and `type` = 'update' " +
                        "and data['refund_status'] = '0705' "
//                "and `old`[refund_status] is not null "
        );

        tenv.createTemporaryView("refund_payments",refundPayment);


        // TODO 6. 读取订单表数据并过滤退款成功订单数据

        Table orderInfo = tenv.sqlQuery("select " +
                "data['id'] id, " +
                "data['user_id'] user_id, " +
                "data['province_id'] province_id, " +
                "`old` " +
                "from topic_db " +
                "where `table` = 'order_info'" +
                "and `type` = 'update'" +
                "and data['order_status'] = '1006'" +
                "and `old`['order_status'] is not null ");
        tenv.createTemporaryView("order_info",orderInfo);


        //TODO 7. 读取退单表数据并过滤退款成功数据
        Table orderRefundInfo = tenv.sqlQuery("select " +
                "data['order_id'] order_id, " +
                "data['sku_id'] sku_id, " +
                "data['refund_num'] refund_num, " +
                "`old` " +
                "from topic_db " +
                "where `table` = 'order_refund_info'" +
                "and `type` = 'update'" +
                "and data['refund_status'] = '0705'"
//                "and `old`['refund_status'] is not null "
        );
        tenv.createTemporaryView("order_refund_info",orderRefundInfo);

        // TODO 8. 关联四张表获得退款成功表
        Table resTable = tenv.sqlQuery("select " +
                "rp.id, " +
                "rp.order_id, " +
                "rp.sku_id, " +
                "rp.payment_type, " +
                "rp.callback_time, " +
                "rp.total_amount, " +
                "date_format(rp.callback_time,'yyyy_MM_dd') data_id, " +
                "oi.user_id, " +
                "oi.province_id, " +
                "ri.refund_num, " +
                "dic.dic_name payment_type_name, " +
                "rp.ts, " +
                "current_row_timestamp() row_op_ts " +
                "from refund_payments rp " +
                "join " +
                "order_info oi " +
                "on rp.order_id = oi.id " +
                "join " +
                "order_refund_info ri " +
                "on ri.order_id = rp.order_id " +
                "and rp.sku_id = ri.sku_id " +
                "join " +
                "base_dic for system_time as of rp.proc_time as dic " +
                "on rp.payment_type = dic.dic_code");
        tenv.createTemporaryView("res_table",resTable);


        tenv.executeSql("create table dwd_trade_refund_pay_suc(" +
                "id string, " +
                "order_id string, " +
                "sku_id string, " +
                "payment_type string, " +
                "callback_time string, " +
                "total_amount string, " +
                "data_id string, " +
                "user_id string, " +
                "province_id string, " +
                "refund_num string, " +
                "payment_type_name string, " +
                "ts string, " +
                "row_op_ts timestamp_ltz(3) " +
                ")" + MyKafkaUtil.getKafkaSinkDDL("dwd_trade_refund_pay_suc"));


        tenv.executeSql("insert into dwd_trade_refund_pay_suc select * from res_table");
    }
}
